/*
 *  Copyright 2013 National Institute of Advanced Industrial Science and Technology
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.sss.mapreduce.example;

import java.io.StringWriter;

import org.json.simple.JSONObject;
import org.slf4j.Logger;
import org.sss.mapreduce.Output;
import org.sss.mapreduce.Reducer;
import org.sss.mapreduce.SssException;
import org.sss.mapreduce.datatype.DCPackableString;
import org.sss.mapreduce.datatype.PackableInt;
import org.sss.mapreduce.datatype.PackableJSONObject;
import org.sss.mapreduce.datatype.PackableString;
import org.sss.server.SssServer;

public class ContApacheLogReducer extends Reducer {
  private final Logger logger = SssServer.getLogger(ContApacheLogReducer.class);
  private int scheduled = 0;
  @Override
  public void configure(Context context) throws Exception {
    this.scheduled = 0;
  }
  @SuppressWarnings("unchecked")
  public void reduce(Context context,
                     DCPackableString key, Iterable<PackableInt> values,
                     Output<PackableString, PackableJSONObject> output)
    throws Exception {
    logger.info("reduce method");
    int sum = 0;
    for (PackableInt v: values) {
      sum += v.get();
    }
    JSONObject obj = new JSONObject();
    obj.put("sum", new Integer(sum));
    obj.put("scheduled", new Integer(scheduled));
    StringWriter out = new StringWriter();
    obj.writeJSONString(out);
    String jsonText = out.toString();
    output.write(new PackableString(key.get()), new PackableJSONObject(jsonText));
  }
  public void cleanup(Context context,
                      Output<PackableString, PackableJSONObject> output)
    throws SssException {
    logger.info("cleanup is called");
    scheduled++;
  }
}
